package com.tc.vms.producer;

import com.tc.vms.MessageWrap;
import com.tc.vms.connection.VmsConnectionManager;
import com.tc.vms.exception.VmsClientException;
import com.tc.vms.iface.IContext;
import com.tc.vms.iface.TransactionMessageListener;
import com.tc.vms.kafka.KafkaKeySerializer;
import com.tc.vms.kafka.KafkaMessageSerializer;
import com.tc.vms.kafka.KafkaShardAlgorithm;
import com.tc.vms.log.VmsClientLogger;
import com.tc.vms.message.SendResultType;
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
import org.slf4j.Logger;

import java.util.Map;
import java.util.Properties;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Created by yonghua.zhang on 2015/12/14.
 */
public final class VmsProducerKafkaImpl extends AbstractPublisher {
    protected final static Logger LOGGER = VmsClientLogger.getLog();
    private Producer<String, MessageWrap> scalaProducer;
    private String pubTopic;
    private final AtomicInteger keyGenerator = new AtomicInteger(0);

    public VmsProducerKafkaImpl(VmsProducerProxy vmsProducerProxy, VmsConnectionManager vmsConnectionManager) {
        super(vmsProducerProxy, vmsConnectionManager);
    }

    private Properties prepareKafkaProperties() {
        ChannelMetaData channelMetaData = vmsProducerProxy.getVmsMetaData();
        String hosts = channelMetaData.getBrokerCluster();
        pubTopic = channelMetaData.getRealChlNmOnBroker();
        String pubInstName = channelMetaData.getProducerInstName();
        Map<?, ?> userProps = vmsProducerProxy.getUserProps();

        Properties properties = new Properties();
        for (Map.Entry entry : userProps.entrySet()) {
            properties.put(entry.getKey(), entry.getValue());
        }

        properties.put("client.id", pubInstName);
        properties.put("metadata.broker.list", hosts);

        if (properties.getProperty("serializer.class") == null) {
            properties.put("serializer.class", KafkaMessageSerializer.class.getName());
        }
        if (properties.getProperty("key.serializer.class") == null) {
            properties.put("key.serializer.class", KafkaKeySerializer.class.getName());
        }
        if (properties.getProperty("partitioner.class") == null) {
            properties.put("partitioner.class", KafkaShardAlgorithm.class.getName());
        }

        if (serviceSLA != null) {
            properties.put("message.send.max.retries", String.valueOf(serviceSLA.retryTimesWhenSend()));
            if (properties.contains("producer.type")) {
                LOGGER.warn("User properties include Send-Mode parameters[{}], vms will replace it by SLA's parameters [{}]",
                        properties.getProperty("producer.type"),
                        serviceSLA.sendMode().getName());
            }

            if (serviceSLA.isFailFast()) {
                properties.put("retry.backoff.ms", "0");

            }

            switch (serviceSLA.sendMode()) {
                case SYNC:
                    properties.put("producer.type", "sync");
                    properties.put("request.required.acks", "1");

                    properties.put("request.timeout.ms", String.valueOf(serviceSLA.confirmTimeoutInMs()));
                    break;
                case ASYNC:
                    properties.put("producer.type", "async");
                    properties.put("request.required.acks", "1");
                    properties.put("queue.enqueue.timeout.ms", "-1");
                    properties.put("batch.num.messages", String.valueOf(serviceSLA.batchCount()));
                    properties.put("queue.buffering.max.messages", String.valueOf(serviceSLA.asyncQueueSize()));
                    break;
                case ONEWAY:
                    properties.put("producer.type", "async");
                    properties.put("queue.enqueue.timeout.ms", "0");
                    properties.put("request.required.acks", "0");
                    break;
            }

            switch (serviceSLA.msgMode()) { //reserve it for use in the future
                case SEQ_MESG:
                    if (properties.get("partitioner.class") == null) {
                        properties.put("partitioner.class", KafkaShardAlgorithm.class.getName());
                    }
                    break;
                case TRANSACT_MESG:
                    LOGGER.warn("Kafka couldn't support the mode of TRANSACT_MESG, replace by NORMAL_MESG");
                case NORMAL_MESG:
                    break;
            }

        }
        return properties;
    }

    @Override
    public synchronized boolean init(IContext context) throws VmsClientException {
        if (scalaProducer == null) {
            try {
                Properties prepared = prepareKafkaProperties();
                scalaProducer = new Producer<String, MessageWrap>(new ProducerConfig(prepared));
            } catch (Exception ex) {
                LOGGER.error("VmsProducerKafkaImpl init failed.", ex);
                throw new VmsClientException("VmsProducerKafkaImpl init failed.", ex);
            }
        }
        return true;
    }

    @Override
    public boolean start() throws VmsClientException {
        if (!isOpen) {
            isOpen = init(null);
        }
        return isOpen;
    }

    @Override
    public void shutdown() {
        if (isOpen) {
            isOpen = false;
            if (scalaProducer != null) {
                try {
                    scalaProducer.close();
                } catch (Exception e) {
                }
                scalaProducer = null;
            }
        }
    }

    @Override
    public SendResult sendMessage(MessageWrap message) throws VmsClientException {
        return this.sendMessageByOrder(message, null);
    }

    @Override
    public SendResult sendMessageByOrder(MessageWrap message, String partKey) throws VmsClientException {
        try {
            if (partKey == null) {
                int seqKey = keyGenerator.incrementAndGet();
                if (seqKey > Short.MAX_VALUE) {
                    seqKey = 0;
                    keyGenerator.set(seqKey);
                }
                partKey = String.valueOf(seqKey);
            }
            KeyedMessage<String, MessageWrap> keyedMessage = new KeyedMessage<String, MessageWrap>(pubTopic, partKey, message);
            scalaProducer.send(keyedMessage);
            return new SendResult(SendResultType.VMS_SEND_OK, message.getMessageId(), null);
        } catch (Exception ex) {
            LOGGER.debug("Kafka send message error.", ex);
            return new SendResult(SendResultType.VMS_SEND_FAILED, message.getMessageId(), ex.getMessage());
        }
    }

    @Override
    public SendResult sendTransactionMessage(MessageWrap message, TransactionMessageListener listener) throws VmsClientException {
        throw new VmsClientException("Don't support sending transaction message.");
    }

}
